package com.pine.dcts.config;

import com.alibaba.datax.common.constant.PluginType;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.core.util.ConfigParser;
import com.alibaba.datax.core.util.FrameworkErrorCode;
import com.alibaba.datax.core.util.HttpClientUtil;
import com.alibaba.datax.core.util.container.CoreConstant;
import com.pine.dcts.utils.LoaderUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.http.client.methods.HttpGet;

import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import static com.pine.dcts.utils.FileUtils.getAbsPath;


@Slf4j
public class DctsConfigParser {

    public static final String  DATAX_PLUGIN_READER_HOME="plugin/reader";

    public static final String  DATAX_PLUGIN_WRITER_HOME="plugin/writer";

    /**
     * 指定Job配置路径，ConfigParser会解析Job、Plugin、Core全部信息，并以Configuration返回
     */
    public static Configuration parse(String json) {

        Configuration configuration = Configuration.from(json);

//        configuration.merge(
//                ConfigParser.parseCoreConfig(CoreConstant.DATAX_CONF_PATH),
//                false);
        // todo config优化，只捕获需要的plugin
        String readerPluginName = configuration.getString(
                CoreConstant.DATAX_JOB_CONTENT_READER_NAME);
        String writerPluginName = configuration.getString(
                CoreConstant.DATAX_JOB_CONTENT_WRITER_NAME);

        String preHandlerName = configuration.getString(
                CoreConstant.DATAX_JOB_PREHANDLER_PLUGINNAME);

        String postHandlerName = configuration.getString(
                CoreConstant.DATAX_JOB_POSTHANDLER_PLUGINNAME);

        Set<String> pluginList = new HashSet<String>();
        pluginList.add(LoaderUtils.generatePluginKey(PluginType.READER,readerPluginName));
        pluginList.add(LoaderUtils.generatePluginKey(PluginType.WRITER,writerPluginName));
      //  pluginList.add(writerPluginName);

        if(StringUtils.isNotEmpty(preHandlerName)) {
           // pluginList.add(preHandlerName);
            pluginList.add(LoaderUtils.generatePluginKey(PluginType.HANDLER,preHandlerName));
        }
        if(StringUtils.isNotEmpty(postHandlerName)) {
           // pluginList.add(postHandlerName);
            pluginList.add(LoaderUtils.generatePluginKey(PluginType.HANDLER,postHandlerName));
        }
        try {
            configuration.merge(parsePluginConfig(new ArrayList<>(pluginList)), false);
        }catch (Exception e){
            //吞掉异常，保持log干净。这里message足够。
            log.warn(String.format("插件[%s,%s]加载失败，1s后重试... Exception:%s ", readerPluginName, writerPluginName, e.getMessage()));
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e1) {
                //
            }
            configuration.merge(parsePluginConfig(new ArrayList<String>(pluginList)), false);
        }

        return configuration;
    }


    public static Configuration parsePluginConfig(List<String> wantPluginNames ){
        Configuration configuration = Configuration.newDefault();
        int complete = 0;
        for(String key:wantPluginNames){
            Configuration temp  = PluginManager.pluginRegisterCenter.get(key);
            if(temp!=null){
                configuration.merge(temp,true);
                complete++;
            }
        }
        if (wantPluginNames != null && wantPluginNames.size() > 0 && wantPluginNames.size() != complete) {
            throw DataXException.asDataXException(FrameworkErrorCode.PLUGIN_INIT_ERROR, "插件加载失败，未完成指定插件加载:" + wantPluginNames);
        }
        return configuration;
    }

    public static Configuration parseCoreConfig(String path) {
        return Configuration.from(new File(path));
    }


    public static String getJobContent(String jobResource) {
        String jobContent;

        boolean isJobResourceFromHttp = jobResource.trim().toLowerCase().startsWith("http");


        if (isJobResourceFromHttp) {
            //设置httpclient的 HTTP_TIMEOUT_INMILLIONSECONDS
            Configuration coreConfig = Configuration.from(new File(CoreConstant.DATAX_CONF_PATH));
            int httpTimeOutInMillionSeconds = coreConfig.getInt(
                    CoreConstant.DATAX_CORE_DATAXSERVER_TIMEOUT, 5000);
            HttpClientUtil.setHttpTimeoutInMillionSeconds(httpTimeOutInMillionSeconds);

            HttpClientUtil httpClientUtil = new HttpClientUtil();
            try {
                URL url = new URL(jobResource);
                HttpGet httpGet = HttpClientUtil.getGetRequest();
                httpGet.setURI(url.toURI());

                jobContent = httpClientUtil.executeAndGetWithFailedRetry(httpGet, 1, 1000l);
            } catch (Exception e) {
                throw DataXException.asDataXException(FrameworkErrorCode.CONFIG_ERROR, "获取作业配置信息失败:" + jobResource, e);
            }
        } else {
            // jobResource 是本地文件绝对路径
            try {
                jobContent = FileUtils.readFileToString(new File(jobResource));
            } catch (IOException e) {
                throw DataXException.asDataXException(FrameworkErrorCode.CONFIG_ERROR, "获取作业配置信息失败:" + jobResource, e);
            }
        }

        if (jobContent == null) {
            throw DataXException.asDataXException(FrameworkErrorCode.CONFIG_ERROR, "获取作业配置信息失败:" + jobResource);
        }
        return jobContent;
    }

//    public static Configuration parsePluginConfig(List<String> wantPluginNames) {
//        Configuration configuration = Configuration.newDefault();
//
//        Set<String> replicaCheckPluginSet = new HashSet<>();
//        int complete = 0;
//        for (final String each : getDirAsList(DATAX_PLUGIN_READER_HOME)) {
//            Configuration eachReaderConfig = ConfigParser.parseOnePluginConfig(each, "reader", replicaCheckPluginSet, wantPluginNames);
//            if(eachReaderConfig!=null) {
//                configuration.merge(eachReaderConfig, true);
//                complete += 1;
//            }
//        }
//
//        for (final String each : getDirAsList(DATAX_PLUGIN_WRITER_HOME)) {
//            Configuration eachWriterConfig = ConfigParser.parseOnePluginConfig(each, "writer", replicaCheckPluginSet, wantPluginNames);
//            if(eachWriterConfig!=null) {
//                configuration.merge(eachWriterConfig, true);
//                complete += 1;
//            }
//        }
//
//        if (wantPluginNames != null && wantPluginNames.size() > 0 && wantPluginNames.size() != complete) {
//            throw DataXException.asDataXException(FrameworkErrorCode.PLUGIN_INIT_ERROR, "插件加载失败，未完成指定插件加载:" + wantPluginNames);
//        }
//
//        return configuration;
//    }

    private static List<String> getDirAsList(String path) {
        path =getAbsPath(path);
        List<String> result = new ArrayList<>();
        String[] paths = new File(path).list();
        if (null == paths) {
            return result;
        }

        for (final String each : paths) {
            File file = new File(StringUtils.join(
                    new String[] { path, each }, File.separator));
            if(file.exists()&&file.isDirectory())
            result.add(path + File.separator + each);
        }
        return result;
    }


}
